package hyl.base.zk;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;

import hyl.core.net.MySocket;

/**
 * zkapi是基础工具类
 * 
 * @author 37798955@qq.com
 *
 */
public class MyZK implements Watcher {

	ZooKeeper zk;
	// 阻塞用户线程，用户必须等待连接成功
	private CountDownLatch countDownLatch = new CountDownLatch(1);

	static MyZK _zk = null;

	public static MyZK getInstance(String address, int timeout) throws IOException, InterruptedException {
		if (_zk == null)
			_zk = new MyZK(address, timeout);

		return _zk;
	}

	public MyZK(String address, int timeout) throws IOException, InterruptedException {
		connect(address, timeout);
	}

	void connect(String address, int timeout) throws IOException, InterruptedException {
		zk = new ZooKeeper(address, timeout, this);	
		countDownLatch.await();// 等待连接成功,否则阻塞		
	}
	public boolean isConnected(String address, int timeout) {
		return MySocket.checkAddress(address, timeout);
		
	}

	/**
	 * 关闭ZK连接
	 */
	public void close() {
		if (this.zk != null) {
			try {
				this.zk.close();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 创建单个节点
	 * 如果节点存在 更新数据
	 * @param path
	 * @param data
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public void setPNode(String path, byte[] data) throws KeeperException, InterruptedException {
		Stat stat = zk.exists(path, false);
		// 父节点不存在，则创建父节点
		if (null == stat) {
			synchronized (path) {// 哪怕是读取也会被锁住
				// 防止这一会时间 节点 又存在了
				Stat stat2 = zk.exists(path, /* needWatch */true);
				if (null == stat2) {
					// 父节点是持久节点
					zk.create(path, data, /** 所有可见 */
							Ids.OPEN_ACL_UNSAFE, /** 永久存储 */
							CreateMode.PERSISTENT);
				}
			}
		} else {
			zk.setData(path, data, stat.getVersion());
		}
	}

	/**
	 * 创建路径上的所有节点
	 * 
	 * @param path
	 * 
	 * @return
	 */
	public void createPNodes(String path) {
		// 第三个参数表示权限的，这里开放所有权限，不限制服务器
		// 第四个参数表示节点的类型。用的持久节点
		createPNodes(path, null);
	}

	/**
	 * 创建路径上的所有节点
	 * 
	 * @param path 完整路径
	 * @param data 设置叶结点的值
	 */
	public void createPNodes(String path, byte[] data) {
		// 第三个参数表示权限的，这里开放所有权限，不限制服务器
		// 第四个参数表示节点的类型。用的持久节点
		try {
			// 开启监听的方法。第二个参数表示是否开启监听

			path = path.replace("\\", "/").replace("//", "/");
			if (path.startsWith("/"))
				path = path.substring(1);
			if (path.endsWith("/"))
				path = path.substring(0, path.length() - 1);
			StringBuilder sb = new StringBuilder();
			String ss[] = path.split("/");
			if (ss.length == 0)
				return;
			int max = ss.length - 1;
			for (int i = 0; i < max; i++) {
				if (ss[i].isEmpty())
					continue;
				// System.out.println(str+"--");
				sb.append("/").append(ss[i]);
				setPNode(sb.toString(), null);
			}
			sb.append("/").append(ss[max]);
			setPNode(sb.toString(), data);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * 创建临时节点<br>
	 * 临时节点下面不能创建临时节点
	 * 
	 * @param path 不能是临时节点路径
	 * @param data
	 * @return
	 */
	public String createTNode(String path, byte[] data) {
		try {
			// 开启监听的方法。第二个参数表示是否开启监听
			return zk.create(path + "/", data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}


	/**
	 * 判断指定节点是否存在
	 * 
	 * @param path 节点路径
	 */
	public Stat exists(String path, Watcher watch) {
		try {
		  return zk.exists(path, watch);
		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}

	public boolean exists(String path) {
		try {
			Stat st = zk.exists(path, false);
			if (st == null)
				return false;
			else
				return true;
		} catch (KeeperException | InterruptedException e) {
			return false;
		}

	}

	/**
	 * 读取指定节点数据内容
	 * 
	 * @param path 节点路径
	 * @return
	 */
	public byte[] getData(String path, Watcher needWatch) {
		try {

			return this.zk.getData(path, needWatch, null);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * 读取指定节点数据内容
	 * 
	 * @param path 节点路径
	 * @return
	 */
	public byte[] getData(String path, boolean needWatch) {
		try {

			return this.zk.getData(path, needWatch, null);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * 更新指定节点数据内容
	 * 
	 * @param path 节点路径
	 * @param data 数据内容
	 * @return
	 */

	public Stat setData(String path, String data) {
		// 第三个参数表示权限的，这里开放所有权限，不限制服务器
		// 第四个参数表示节点的类型。用的持久节点
		try {
			// 开启监听的方法。第二个参数表示是否开启监听
			Stat st = zk.exists(path, true);
			if (st == null)
				return null;
			// 第三个参数表示版本号。 zk的数据版本默认从0开始，每次修改都会加1. -1严格来说属于不合法的版本号。表示从最新版本进行更新
			return zk.setData(path, data.getBytes(), -1);
			// System.out.println("修改节点成功！节点为："+path+",修改后为值为:"+data);
			// System.out.println("修改节点成功，result:"+result);

		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	public void deleteChildren(String path) {
		List<String> cds = getChildren(path);
		if (null == cds)
			return;
		for (String s : cds) {
			deleteNode(path + '/' + s);
		}
	}

	public List<String> getChildren(String path) {
		try {
			if (null != zk.exists(path, false)) {
				List<String> cds = zk.getChildren(path, false);
				return cds;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 删除自身或子节点或创建子节点时触发
	 * 
	 * @param path
	 * @param 监听器
	 * @return
	 */
	public List<String> watchChildren(String path, Watcher 监听器) {

		try {
			if (null != zk.exists(path, false)) {
				return zk.getChildren(path, 监听器);

			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;

	}

	public void deleteNode(String path) {
		// 第二个参数表示版本号
		try {
			// 开启监听的方法。第二个参数表示是否开启监听
			if (zk.exists(path, true) != null) {
				// 第二个参数表示版本号。 zk的数据版本默认从0开始，每次修改都会加1. -1严格来说属于不合法的版本号。表示从最新版本进行更新
				zk.delete(path, -1);
			}
		} catch (Exception e) {
			e.printStackTrace();

		}
	}

	/**
	 * 
	 */
	public void process(WatchedEvent event) {
		// System.out.println("事件通知开始前");
		// 事件路径
		// String path = event.getPath();
		// 事件状态 即 连接不连接
		KeeperState state = event.getState();
		// 事件类型
		EventType type = event.getType();
		// System.out.println("事件路径"+path+",事件状态"+state+",事件类型"+type);
		if (KeeperState.SyncConnected == state) {
			// 事件类型 None表示是连接类型
			if (EventType.None == type) {
				// System.out.println("连接类型");
				// 连接上zk服务器后放开信号量
				countDownLatch.countDown();
			} /*
				 * System.out.println("=====ZK连接成功====="); } else if (EventType.NodeCreated ==
				 * type) { // System.out.println("=====新增节点成功=====path:" + path); } else if
				 * (EventType.NodeDataChanged == type) { //
				 * System.out.println("=====修改节点成功=====path:" + path); } else if
				 * (EventType.NodeDeleted == type) { //
				 * System.out.println("=====删除节点成功=====path:" + path); }
				 */
		}
		// System.out.println("事件通知开始后");
	}

	public static void main(String[] args) throws IOException, InterruptedException {
		MyZK w = new MyZK("192.168.0.113:2181", 1000);

		w.createPNodes("/zk100/zk110/zk111");
		w.createTNode("/zk100/zk110/zk111", "zk01-value".getBytes());
		// w.updateNode("/zk01", "zk01-value2");
		w.deleteChildren("/zk100/zk110/zk111");
		w.deleteNode("/zk100/zk110/zk111");

		w.deleteChildren("/zk100/zk110");
		w.deleteNode("/zk100/zk110");

		w.deleteChildren("/zk100");
		w.deleteNode("/zk100");
	}
}
